package com.insight.common.mq.aspect;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.AfterThrowing;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.jeecg.common.mq.AckFlusher;
import org.jeecg.common.mq.DefaultRedisMq;
import org.jeecg.common.mq.aspect.annotation.RedisMessageExtend;
import org.jeecg.common.mq.stream.RedisMqStream;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.util.Map;

/**
 * @author JinPengDong
 * @date 2023/10/11
 */
@Slf4j
@Aspect
@Component
public class RedisMessageExtendAspect {

    @Pointcut("@annotation(org.jeecg.common.mq.aspect.annotation.RedisMessageExtend)")
    public void mqPointCut() {

    }

    @Before("mqPointCut()")
    public void before(JoinPoint point) {
        final JSONObject messageInfo = getMessageInfo(point);
        if (messageInfo == null) {
            log.warn("getMessageInfo 返回 null");
            return;
        }

        String streamName = messageInfo.getString("streamName");
        String consumerGroup = messageInfo.getString("consumerGroup");
        String content = messageInfo.getString("content");
        String recordId = messageInfo.getString("recordId");
        //消费消息
        log.info("消费消息,streamName:{},consumerGroup:{}", streamName, consumerGroup);
        if (log.isDebugEnabled()) {
            log.debug("消费消息,streamName:{},consumerGroup:{},data:{}", streamName, consumerGroup, content);
        }
        //执行方法
        try {
            String key = String.format("%s^_^%s", consumerGroup, streamName);
            AckFlusher ackFlusher = DefaultRedisMq.consumerMap.get(key);
            if (ackFlusher != null) {
                ackFlusher.addAckId(recordId);
            } else {
                log.warn("AckFlusher 不存在，key: {}", key);
            }
        } catch (Exception e) {
            log.error("登记确认 ID 失败", e);
        }
    }

    @AfterThrowing(value = "mqPointCut()", throwing = "ex")
    public void afterThrowing(JoinPoint point, Exception ex) {
        MethodSignature signature = (MethodSignature) point.getSignature();
        final RedisMqStream redisMqStream = signature.getMethod().getAnnotation(RedisMessageExtend.class).stream().getValue();
        log.error("streamName:{},consumerGroup:{}", redisMqStream.getName(), redisMqStream.getGroup());
        log.error(ex.getMessage(), ex);
    }

    private JSONObject getMessageInfo(JoinPoint point) {
        final MapRecord<String, String, String> message = (MapRecord<String, String, String>) point.getArgs()[0];
        //获取
        MethodSignature signature = (MethodSignature) point.getSignature();
        final RedisMqStream redisMqStream = signature.getMethod().getAnnotation(RedisMessageExtend.class).stream().getValue();
        RecordId id = message.getId();
        final Map<String, String> value = message.getValue();
        JSONObject messageInfo = new JSONObject();
        messageInfo.put("streamName", redisMqStream.getName());
        messageInfo.put("consumerGroup", redisMqStream.getGroup());
        messageInfo.put("recordId", id.getValue());
        messageInfo.put("content", value.get("content"));
        return messageInfo;
    }
}
